package com.dgz.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2025/08/21/08:38
 * @Description:
 */
//无界 流
public class WordCountStreamUnboundeDemo {

    public static void main(String[] args) throws Exception {
//        创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        读取数据
        DataStreamSource<String> socketDs = env.socketTextStream("hadoop102", 7777);
//        处理数据
        socketDs.flatMap((String value, Collector<Tuple2<String, Integer>> collector)->{
            String[] words = value.split(" ");
            for (String word : words) {
                collector.collect(Tuple2.of(word, 1));
            }
        })
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(stringIntegerTuple2 -> stringIntegerTuple2.f0)
                .sum(1).print();
//        输出
//        执行
        env.execute();
    }
}
